Fix: rd_kafka_consume_batch incorrectly advances offset by 2 on EOF messages #5213
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…a_q_serve_rkmessages function
|
🎉 All Contributor License Agreements have been signed. Ready to merge. |
…h_queue This commit introduces a new test, do_test_consume_batch_eof_position, to verify that the consumer position is correctly updated when EOF messages are received with enable.partition.eof set to true. The test ensures that the consumer position advances by 1 instead of 2, addressing a previously identified bug. The test includes setup for producing messages, consuming them, and validating the final consumer position after EOF is reached.
There was a problem hiding this comment.
Pull Request Overview
This PR fixes a bug in rd_kafka_consume_batch() where EOF messages incorrectly advance the consumer position by 2 instead of 1 when enable.partition.eof=true, causing rd_kafka_position() to return an incorrect offset.
- Added a check to prevent EOF messages from updating consumer position in
rd_kafka_q_serve_rkmessages() - Added comprehensive regression test to verify correct consumer position after EOF messages
Reviewed Changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| src/rdkafka_queue.c | Fixed offset advancement logic to exclude EOF messages from position updates |
| tests/0137-barrier_batch_consume.c | Added regression test to verify EOF message position handling |
Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.
This commit modifies the comment in the do_test_consume_batch_eof_position function to clarify that the position value is being extracted from the partition list after reaching EOF. This change enhances code readability and understanding of the test's purpose.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary
This PR fixes a bug in
rd_kafka_consume_batch()where EOF messages incorrectly advance the consumer position by 2 instead of 1 whenenable.partition.eof=true. The bug causesrd_kafka_position()to return last_offset + 2 instead of the correct last_offset + 1 after consuming an EOF message in batch mode.Motivation
Root Cause
In
src/rdkafka_queue.c, therd_kafka_q_serve_rkmessages()function updates the consumer position for all messages in a batch, including EOF messages. EOF messages have error codeRD_KAFKA_RESP_ERR__PARTITION_EOFbut are not control messages, so they were incorrectly incrementing the offset.Changes
File:
src/rdkafka_queue.cAdded a check to prevent EOF messages from updating the consumer position.
Regression subtest:
tests/0137-barrier_batch_consume.cNew Subtest
do_test_consume_batch_eof_position()is addedTest Flow
rd_kafka_consume_batch_queue()withenable.partition.eof=truerd_kafka_position()returnslast_offset + 1(notlast_offset + 2)